// =============================================================
// CCubeMethods interface
// ---------------------------------------
// Author : Woo-Hyun Kim
// Email : woorung@gmail.com or woorung@nhncorp.com
// Date : Feb. 7, 2007
//----------------------------------------------------------------------
#include "CInfoMap.h"
#include "CCubeMethods.h"
#include "CShellProcess.h"
//----------------------------------------------------------------------
CCubeMethods::CCubeMethods(string &name) : CMethod(name)
{
}
//----------------------------------------------------------------------
CCubeMethods::~CCubeMethods()
{
}
//----------------------------------------------------------------------
// bind the requested calls to real methods
//----------------------------------------------------------------------
void CCubeMethods::run() {
	if(call_name=="parallelization") parallelProcessing(call_name, params);
	else if(call_name=="distribution") parallelDistribution(call_name, params);
	else if(call_name=="aggregation") doAggregation(call_name, params);
	else {
		cout << "///////////////////////////////////////////////////////////" << endl;
		cout << "[error] : Not found such an invoked method : " << call_name << endl;
		cout << "///////////////////////////////////////////////////////////" << endl;
		exit(-1);
	}
}
//----------------------------------------------------------------------
// common functions
//----------------------------------------------------------------------
void CCubeMethods::inspectParameters(string &name, map<string,string> &parameters)
{
	cout << endl;
    cout << "[inspection] : " << name << endl;
	cout << "----------------------------------------------------------" << endl;

    map<string,string>::iterator param = parameters.begin();
    while(param!=parameters.end()) {
        cout << param->first << "=" << param->second << endl;

        param++;
    }
	cout << "----------------------------------------------------------" << endl << endl;
}
//----------------------------------------------------------------------
void CCubeMethods::generateDirectory(string &default_path, string &current_path)
{
    struct stat statbuf;
    struct dirent *dirp;
    DIR *dp;

    string curdir = default_path;
    string curpath = current_path;
    string delim = "/";

	int start = curpath.find(curdir);
	int len = curdir.size();
	if(start>=0&&start<len) curpath = curpath.substr(start+len);

    list<string> dirs;
    CSmartString *sm = new CSmartString(curpath, delim);
    sm->CSplit(dirs);
    delete sm;

    list<string>::iterator current = dirs.begin();
    while(current!=dirs.end()) {
        curdir += "/";
        curdir += *current;
        if((dp=opendir(curdir.c_str()))==NULL) {
            mkdir(curdir.c_str(),07777);
            cout << "[generate] : " << curdir << endl;
        }
        if(dp!=NULL) {
            //cout << "[already exist] : " << curdir << endl;
            closedir(dp);
        }

        current++;
    }
}
//----------------------------------------------------------------------
void CCubeMethods::removeDirectory(string &target_path)
{
	DIR *dp;
	struct dirent *ep;
	if((dp = opendir(target_path.c_str()))==NULL) {
		while(ep=readdir(dp)) {
			string d_name = ep->d_name;
			remove(d_name.c_str());
		}
	}
	if(dp!=NULL) {
		closedir(dp);
	}
}
//----------------------------------------------------------------------
// common methods for cube
//----------------------------------------------------------------------
void CCubeMethods::initializeEnvironment(string &name, map<string,string> &parameters)
{
	if(parameters["cube"]=="") {
		inspectParameters(name, parameters);

		cout << "[environment] error : @cube must be set" << endl;
		exit(-1);
	}

	string cube_zero = "";

	generateDirectory(cube_zero, parameters["cube"]);

	string delim = " ";
	list<string> cube_names;
	CSmartString *cube_sm = new CSmartString(parameters["available_cubes"], delim);
	cube_sm->CSplit(cube_names);
	delete cube_sm;

	char tmp[1024];
	list<string>::iterator cube_name = cube_names.begin();
	while(cube_name!=cube_names.end()) {
		string cname = *cube_name;
		int num = atoi(cname.c_str());

		//sprintf(tmp,"%s%03da",LOCAL_CUBE_PATH,num);
		sprintf(tmp,"%s%03d",LOCAL_CUBE_PATH,num);
		cubes[cname] = tmp;

		generateDirectory(cubes[cname], parameters["cube_base"]);
		generateDirectory(cubes[cname], parameters["local_dataset"]);
		generateDirectory(cubes[cname], parameters["local_resultset"]);

		cube_name++;
	}

    //inspectParameters(name, parameters);
}
//----------------------------------------------------------------------
void CCubeMethods::parallelProcessing(string &name, map<string,string> &parameters)
{
	initializeEnvironment(name, parameters);

	string pure_cmd = parameters["run"];

	if(pure_cmd=="") {
		inspectParameters(name, parameters);

		cout << "[parallelization] error : @run must be set" << endl;
		exit(-1);
	}

	vector<CShellProcess> parallel_cmds;

	// get cube list
	string delim = " \t\r\n";
	list<string> cube_names;
	CSmartString *cube_sm = new CSmartString(parameters["available_cubes"], delim);
	cube_sm->CSplit(cube_names);
	delete cube_sm;

	// get dynamic parameters list
	list<string> dparam_names;
	if(parameters["dynamic_parameters_set"]!="") {
		CSmartString *dparam_sm = new CSmartString(parameters["dynamic_parameters_set"], delim);
		dparam_sm->CSplit(dparam_names);
		delete dparam_sm;
	}

	// make up parallel commands
	string parallel_cmd;
	list<string>::iterator cube_name = cube_names.begin();
	for(int i=0;cube_name!=cube_names.end();i++) {
		string cname = *cube_name;

                parallel_cmd = "onnode " + cname + " ";

                int start = 0;
                int end = pure_cmd.length();
                while(start>=0&&start<end) {
                        int found = pure_cmd.find(parameters["cube_base"], start);

                        if(found<0||found>=end) {
                                parallel_cmd += pure_cmd.substr(start);
                                break;
                        }
                        else {
                                string noop = pure_cmd.substr(start, found-start);
                                int huck = noop.rfind(LOCAL_CUBE_PATH);

                                parallel_cmd += noop;
                                if(huck<0||huck>=noop.length()||noop.length()-huck>sizeof(LOCAL_CUBE_PATH)) parallel_cmd += cubes[cname];
                                parallel_cmd += pure_cmd.substr(found, parameters["cube_base"].length());

                                start = found + parameters["cube_base"].length();
                        }
                }

		string adjust_cmd = parallel_cmd;

		// make up dynamic parameters
		if(dparam_names.size()>0) {
			string pattern = parameters["dynamic_parameters"];
			CSmartString *each_dparam_sm = new CSmartString(adjust_cmd, pattern);
			while(each_dparam_sm->GetPatternCount()>0) {
				list<string>::iterator dparam = dparam_names.begin();
				each_dparam_sm->SequentialReplace(*dparam);
				string result;
				dparam_names.pop_front();
			}
			adjust_cmd = each_dparam_sm->GetString(adjust_cmd);
			delete each_dparam_sm;
		}

		parallel_cmd = adjust_cmd;

		// vector insert
		parallel_cmds.push_back(CShellProcess(parallel_cmd));

		cube_name++;
	}

	// concurrent execution
	vector<CShellProcess>::iterator current = parallel_cmds.begin();
	while(current!=parallel_cmds.end()) {
		current->start();
		cout << "[execute] : " << current->get() << endl;
		cout << endl;

		current++;
	}

	// wait until commands are finished
	vector<CShellProcess>::iterator finished = parallel_cmds.begin();
	while(finished!=parallel_cmds.end()) {
		if(finished->waits()<0) {
			cout << finished->get_pid() << " was abnormally exited" << endl;
			exit(-1);
		}
		else {
			cout << "######################################################################" << endl;
			cout << " Depredicated function !!!!!!!!!!!!!!!!!!!!!!!!!!!" << endl;
			cout << endl;
			cout << " the result will be printed out over stdout of the exeucting place" << endl;
			cout << "######################################################################" << endl;
/*
			string cmd_out = finished->getCommandResult();
			if(cmd_out!="") {
				cout << cmd_out << endl;
				cout << endl;
			}
*/
		}

		finished++;
	}
}
//----------------------------------------------------------------------
void CCubeMethods::parallelDistribution(string &name, map<string,string> &parameters)
{
    initializeEnvironment(name, parameters);

	string data_path = parameters["dataset"];
	if(data_path=="") {
		inspectParameters(name, parameters);

		cout << "[distribution] error : @dataset must be set" << endl;
		exit(-1);
	}

	vector<CShellProcess> parallel_cmds;

	// get cube list
	string delim = " \t\r\n";
	list<string> cube_names;
	CSmartString *cube_sm = new CSmartString(parameters["available_cubes"], delim);
	cube_sm->CSplit(cube_names);
	delete cube_sm;

	if(parameters["partitioning_type"]=="FILE") {
		vector<string> dirs;
		vector<string> files;

		// get files
		CFileManager *FileMan = new CFileManager();
		FileMan->StartWalking((char *)data_path.c_str());
		dirs = FileMan->getDirs(dirs);
		files = FileMan->getFiles(files);
		delete FileMan;

		// make up parallel commands
		string parallel_cmd;
		vector<string> cube_ids(cube_names.size());
		copy(cube_names.begin(), cube_names.end(), cube_ids.begin());

		for(int i=0;i<files.size();i++) {
			string cname = cube_ids[i%cube_ids.size()];

			parallel_cmd = "onnode " + cname + " ";
			parallel_cmd += "cp -rf " + files[i] + " " + cubes[cname] + parameters["local_dataset"] + "/";

			// vector insert
			parallel_cmds.push_back(CShellProcess(parallel_cmd));
		}

		// concurrent execution in a specific count
		int ec=0;
		int wc=0;
		for(ec=0,wc=0;ec<parallel_cmds.size();ec++) {
			parallel_cmds[ec].start();
			wc++;

			cout << "[execute] : " << parallel_cmds[ec].get() << endl;
			cout << endl;

			// wait for a specific count of runing processes
			if(wc%MAX_CONCURRENT_CP_COUNT==0) {
				for(;wc>0;wc--) {
					if(parallel_cmds[(ec+1)-wc].waits()<0) {
						cout << parallel_cmds[(ec+1)-wc].get_pid() << " was abnormally exited" << endl;
						exit(-1);
					}
					else {
						cout << "######################################################################" << endl;
						cout << " Depredicated function !!!!!!!!!!!!!!!!!!!!!!!!!!!" << endl;
						cout << endl;
						cout << " the result will be printed out over stdout of the exeucting place" << endl;
						cout << "######################################################################" << endl;
/*
						//cout << parallel_cmds[(ec+1)-wc].get_pid() << " was successfully exited" << endl;
						string cmd_out = parallel_cmds[(ec+1)-wc].getCommandResult();
						if(cmd_out!="") {
							cout << cmd_out << endl;
							cout << endl;
						}
*/
					}
				}
			}
		}

		--ec;
		if(wc>0) {
			for(;wc>0;wc--) {
				if(parallel_cmds[(ec+1)-wc].waits()<0) {
					cout << parallel_cmds[(ec+1)-wc].get_pid() << " was abnormally exited" << endl;
					exit(-1);
				}
				else {
					cout << "######################################################################" << endl;
					cout << " Depredicated function !!!!!!!!!!!!!!!!!!!!!!!!!!!" << endl;
					cout << endl;
					cout << " the result will be printed out over stdout of the exeucting place" << endl;
					cout << "######################################################################" << endl;
/*
					//cout << parallel_cmds[(ec+1)-wc].get_pid() << " was successfully exited" << endl;
					string cmd_out = parallel_cmds[(ec+1)-wc].getCommandResult();
					if(cmd_out!="") {
						cout << cmd_out << endl;
						cout << endl;
					}
*/
				}
			}
		}
	}
	else if(parameters["partitioning_type"]=="DOCUMENT") {
	}
	else if(parameters["partitioning_type"]=="KEYVALUE") {
	}
	else if(parameters["partitioning_type"]=="LINE") {
		vector<string> dirs;
		vector<string> files;

		// get files
		CFileManager *FileMan = new CFileManager();
		FileMan->StartWalking((char *)data_path.c_str());
		dirs = FileMan->getDirs(dirs);
		files = FileMan->getFiles(files);

		// make up parallel commands
		string parallel_cmd;
		vector<string> cube_ids(cube_names.size());
		copy(cube_names.begin(), cube_names.end(), cube_ids.begin());

		for(int i=0;i<files.size();i++) {
			ifstream *pf = new ifstream(files[i].c_str());
			if(pf==NULL||!pf->is_open()) {
				cout << "not open the file : " << files[i] << endl;
			}

			// create files to write
			vector<ofstream *> partitions;
			for(int j=0;j<cube_ids.size();j++) {
				string cname = cube_ids[j%cube_ids.size()];

				int pos = files[i].rfind("/");
				string tfile = cubes[cname] + parameters["local_dataset"] + "/" + files[i].substr(pos+1);
				ofstream *apart = new ofstream(tfile.c_str(), ios::app|ios::out);
				partitions.push_back(apart);

				cout << "[distribution] : writing lines to " << tfile << endl;
			}

			char line[MAX_LINE_LENGTH];
			for(int k=0;!pf->eof();k++) {
				do {
					if(pf->eof()) break;

					pf->clear(); // clear flag bits
					pf->getline(line, MAX_LINE_LENGTH);
					*(partitions[k%cube_ids.size()]) << line;
				} while(pf->gcount()==MAX_LINE_LENGTH-1&&pf->fail());

				if(!pf->eof()) *(partitions[k%cube_ids.size()]) << endl;
			}

			// free and close split files
			for(int j=0;j<partitions.size();j++) {
				partitions[j]->close();
				delete (ofstream *) partitions[j];
			}

			cout << "[finish] " << files[i] << endl;

			// close target file
			if(pf->is_open()) pf->close();
			delete pf;
		}
		
	}
	else {
		cout << "[distribution] error : invalid @partitioning_type = " << parameters["partitioning_type"] << endl;
		exit(-1);
	}
}
//----------------------------------------------------------------------
void CCubeMethods::doAggregation(string &name, map<string,string> &parameters)
{
	initializeEnvironment(name, parameters);

	string dataset_path = parameters["dataset"];
	if(dataset_path=="") {
		inspectParameters(name, parameters);
		cout << "[aggregation] error : @dataset must be set" << endl;
		exit(-1);
	}

	string resultset_path = parameters["resultset"];
	if(resultset_path=="") {
		inspectParameters(name, parameters);
		cout << "[aggregation] error : @resultset must be set" << endl;
		exit(-1);
	}

	// get cube list
	string delim = " \t\r\n";
	list<string> cube_names;
	CSmartString *cube_sm = new CSmartString(parameters["available_cubes"], delim);
	cube_sm->CSplit(cube_names);
	delete cube_sm;

	vector<string> cube_ids(cube_names.size());
	copy(cube_names.begin(), cube_names.end(), cube_ids.begin());

	for(int n=0;n<cube_ids.size();n++) {
		string node_name = cube_ids[n%cube_ids.size()];
		string node_dataset_path = cubes[node_name] + dataset_path;

		vector<string> dirs;
		vector<string> files;

		// get files
		CFileManager *FileMan = new CFileManager();
		FileMan->StartWalking((char *)node_dataset_path.c_str());
		dirs = FileMan->getDirs(dirs);
		files = FileMan->getFiles(files);
		delete FileMan;

		// make up parallel commands
		vector<CShellProcess> parallel_cmds;
		string parallel_cmd;

		for(int i=0;i<files.size();i++) {
			parallel_cmd = "onnode " + node_name + " ";
			parallel_cmd += "cat " + files[i] + " >> " + resultset_path;

cout << "[ready command]: " << parallel_cmd << endl;

			// vector insert
			parallel_cmds.push_back(CShellProcess(parallel_cmd));
		}

		for(int ec=0;ec<parallel_cmds.size();ec++) {
			parallel_cmds[ec].start();

			cout << "[execute] : " << parallel_cmds[ec].get() << endl;
			cout << endl;

			if(parallel_cmds[ec].waits()<0) {
				cout << parallel_cmds[ec].get_pid() << " was abnormally exited" << endl;
				exit(-1);
			}
			else {
				cout << "######################################################################" << endl;
				cout << " Depredicated function !!!!!!!!!!!!!!!!!!!!!!!!!!!" << endl;
				cout << endl;
				cout << " the result will be printed out over stdout of the exeucting place" << endl;
				cout << "######################################################################" << endl;
/*
				string cmd_out = parallel_cmds[ec].getCommandResult();
				if(cmd_out!="") {
					cout << cmd_out << endl;
					cout << endl;
				}
*/
			}
		}

/*
		// concurrent execution in a specific count
                int ec=0;
                int wc=0;
                for(ec=0,wc=0;ec<parallel_cmds.size();ec++) {
                        parallel_cmds[ec].start();
                        wc++;

                        cout << "[execute] : " << parallel_cmds[ec].getCommand() << endl;
                        cout << endl;

                        // wait for a specific count of runing processes
                        if(wc%MAX_CONCURRENT_CP_COUNT==0) {
                                for(;wc>0;wc--) {
                                        if(parallel_cmds[(ec+1)-wc].waits()<0) {
                                                cout << parallel_cmds[(ec+1)-wc].get_pid() << " was abnormally exited" << endl;
                                                exit(-1);
                                        }
                                        else {
                                                //cout << parallel_cmds[(ec+1)-wc].get_pid() << " was successfully exited" << endl;
                                                string cmd_out = parallel_cmds[(ec+1)-wc].getCommandResult();
                                                if(cmd_out!="") {
                                                        cout << cmd_out << endl;
                                                        cout << endl;
                                                }
                                        }
                                }
                        }
                }

                --ec;
                if(wc>0) {
                        for(;wc>0;wc--) {
                                if(parallel_cmds[(ec+1)-wc].waits()<0) {
                                        cout << parallel_cmds[(ec+1)-wc].get_pid() << " was abnormally exited" << endl;
                                        exit(-1);
                                }
                                else {
                                        //cout << parallel_cmds[(ec+1)-wc].get_pid() << " was successfully exited" << endl;
                                        string cmd_out = parallel_cmds[(ec+1)-wc].getCommandResult();
                                        if(cmd_out!="") {
                                                cout << cmd_out << endl;
                                                cout << endl;
                                        }
                                }
                        }
                }
*/

		// directory
		for(int i=0;i<dirs.size();i++) {
		}
	}
}
//----------------------------------------------------------------------
